﻿///////////////////////////////////////////////////////////////////////////////
// File:        SocketManager.cc
// Author:      671643387@qq.com
// Date:        2016年1月5日 下午5:44:12
// Description:
///////////////////////////////////////////////////////////////////////////////

#include <cassert>
#if defined(__LINUX__)
#include <netinet/in.h>
#include <arpa/inet.h>
#endif
#include "SocketManager.hpp"

#define __LOCK(x)   fog::thread::LockGuard<fog::thread::Mutex> guard(x)

namespace fog
{
    namespace net
    {
        SocketManager::SocketManager(SocketFactoryPtr factory /* = SocketFactoryPtr */)
            : state_(kNotReady)
            , address_("0.0.0.0")
            , port_(0)
            , backlog_(0)
            , threads_(0)
            , connIdGen_(0)
            , sListen_(-1)
            , evListen_(NULL)
            , sFactory_(factory)
            , sockThreads_()
            , sockMap_()
            , sockMapLock_()
            , acceptLock_()
        {

        }

        SocketManager::~SocketManager(void)
        {

        }

        bool SocketManager::open(const std::string &addr, short port, int backlog, int threads /* = 0 */)
        {
            assert(state_ == kNotReady);
            bool fOk = false;

            do 
            {
                address_ = addr;
                port_ = port;
                backlog_ = backlog;
                threads_ = threads;

                sListen_ = ::socket(AF_INET, SOCK_STREAM, 0);
                if (sListen_ == -1) break;

                struct sockaddr_in saddr;
                memset(&saddr, 0, sizeof(saddr));
                saddr.sin_family = AF_INET;
#if defined(__WINDOWS__)
                saddr.sin_addr.S_un.S_addr = ::inet_addr(addr.c_str());
#elif defined(__LINUX__)
                saddr.sin_addr.s_addr = ::inet_addr(addr.c_str());
#endif
                saddr.sin_port = ::htons(u_short(port));

                if (::bind(sListen_, (const sockaddr*)&saddr, sizeof(saddr)) < 0)
                    break;

                if (threads == 0) threads = 1;
                for (int x = 0; x < threads; ++x)
                {
                    SocketThreadPtr sockThread(new SocketThread());
                    if (!(sockThread && sockThread->open(threads_ == 0)))
                        goto _error;
                    sockThreads_.push_back(sockThread);
                }

                evListen_ = event_new(sockThreads_[0]->evBase_, 
                    sListen_, EV_READ | EV_PERSIST, &SocketManager::onAccept, this);
                if (!evListen_) break;

                if (0 != event_add(evListen_, NULL))
                    break;

                if (::listen(sListen_, backlog) < 0)
                    break;

                if (!Socket::setnonblocking(sListen_))
                    break;

                if (threads_ == 0)
                    state_ = kReady;
                else
                {
                    for (SocketThreadArray::size_type x = 0; 
                        x < sockThreads_.size(); ++x)
                    {
                        if (!sockThreads_[x]->start())
                            goto _error;
                    }

                    state_ = kRunning;
                }

                fOk = true;
            } while (false);
_error:
            if (!fOk)
            {
                for (SocketThreadArray::size_type x = 0; 
                    x < sockThreads_.size(); ++x)
                {
                    sockThreads_[x]->stop();
                    sockThreads_[x]->join();
                }
                sockThreads_.clear();

                if (evListen_)
                {
                    event_free(evListen_);
                    evListen_ = NULL;
                }

                if (sListen_ != -1)
                {
                    Socket::close(sListen_);
                    sListen_ = -1;
                }
            }
            return fOk;
        }

        void SocketManager::join(void)
        {
            if (threads_ == 0)
            {
                assert(state_ == kReady);
                state_ = kRunning;
                sockThreads_[0]->run();
            }
            else
            {
                assert(state_ == kRunning);
                for (SocketThreadArray::size_type x = 0; 
                    x < sockThreads_.size(); ++x)
                {
                    sockThreads_[x]->join();
                }
            }
            sockThreads_.clear();
            state_ = kExit;
        }

        void SocketManager::close(void)
        {
            assert(state_ == kRunning);

            __LOCK(acceptLock_);
            state_ = kExiting;

            if (evListen_)
            {
                event_free(evListen_);
                evListen_ = NULL;
            }
            if (sListen_ != -1)
            {
                Socket::close(sListen_);
                sListen_ = -1;
            }

            {
                __LOCK(sockMapLock_);
                for (SocketMap::iterator it = sockMap_.begin(); 
                    it != sockMap_.end(); ++it)
                {
                    it->second->doClose();
                    it->second->onClose();
                }
                sockMap_.clear();
            }

            for (SocketThreadArray::size_type x = 0; 
                x < sockThreads_.size(); ++x)
            {
                sockThreads_[x]->stop();
            }
        }

        void SocketManager::doCloseSocket(Socket *s)
        {
            assert(s);
            s->actor_->decLoad();
            __LOCK(sockMapLock_);
            SocketMap::iterator it = sockMap_.find(s->id_);
            if (it != sockMap_.end())
            {
                sockMap_.erase(it);
            }
        }

        SocketThreadPtr &SocketManager::findMostFreeThread(void)
        {
            if (threads_ == 0)
            {
                return sockThreads_[0];
            }
            else
            {
                SocketThreadArray::size_type idx;
                int minLoad = INT_MAX;
                for (SocketThreadArray::size_type x = 0; 
                    x < sockThreads_.size(); ++x)
                {
                    int load = sockThreads_[x]->getLoad();
                    if (load < minLoad)
                    {
                        minLoad = load;
                        idx = x;
                    }
                }
                return sockThreads_[idx];
            }
        }

        void SocketManager::onAccept(evutil_socket_t sock, short event, void* arg)
        {
            assert(arg);
            SocketManager *pMgr = static_cast<SocketManager*>(arg);
            pMgr->doAccept();
        }

        void SocketManager::doAccept(void)
        {
            __LOCK(acceptLock_);
            if (state_ != kRunning)
                return;

            evutil_socket_t newsock = ::accept(sListen_, NULL, NULL);
            if (newsock == -1)
                return;

            SocketPtr sock = sFactory_->create();
            if (sock)
            {
                SocketThreadPtr &freeThread = findMostFreeThread();
                freeThread->incLoad();

                if (sock->open(newsock, freeThread.get()))
                {
                    bool fOk = false;
                    {
                        __LOCK(sockMapLock_);
                        int newId = connIdGen_ + 1;
                        if (sockMap_.find(newId) == sockMap_.end())
                        {
                            fOk = true;
                            ++connIdGen_;
                            sock->id_ = newId;
                            sock->owner_ = this;
                            sockMap_[newId] = sock;
                        }
                    }
                    if (fOk) sock->onOpen();
                    return;
                }

                freeThread->decLoad();
                return;
            }
            Socket::close(newsock);
        }
    }
}